package com.example.window;

import com.example.model.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;


/**
 * Created with IntelliJ IDEA.
 * ClassName: WindowFunction
 * Package: com.example.window
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-23
 * Time: 15:06
 */

public class WindowFunction {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        SingleOutputStreamOperator<WaterSensor> map = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] s = value.split(" ");

                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(s[2]));
                    }
                });

        final KeyedStream<WaterSensor, String> key = map.keyBy(value -> value.getId());


        WindowedStream<WaterSensor, String, TimeWindow> window =
                //滚动处理窗口
                key.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        //窗口函数
        //1.增量聚合函数 reduce 和 aggregate函数 来一条计算一条 窗口触发的时候输出结果
        // 三个参数表示  输入的数据类型 累加的数据类型 和 输出的数据类型 可以看着reduce的通用版
        window.aggregate(new AggregateFunction<WaterSensor, Integer, String>() {
            @Override
            public Integer createAccumulator() {
                //每一次新的窗口都会创建
                System.out.println("创建累加的器的方法");
                //相加水位 vc
                //给vc初始值0
                return 0;
            }

            @Override
            public Integer add(WaterSensor value, Integer accumulator) {
                //累加调用的方法 add
                //accumulator上一次累加的结果
                return value.getVc() + accumulator;
            }

            @Override
            public String getResult(Integer accumulator) {
                //窗口触发时候调用
                return accumulator.toString();
            }

            @Override
            public Integer merge(Integer a, Integer b) {
                //一般不会用到 在会话窗口中使用
                return null;
            }
        }).print();
        
        //2.全窗口函数 process 来了数据不计算 窗口触发的时候 计算并输出结果


        env.execute();
    }
}
